package com.zhitan.ems.app.sse.controller;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import com.influxdb.query.FluxTable;
import com.zhitan.ems.app.iot.service.DeviceTopologyPointService;
import com.zhitan.ems.app.iot.vo.TopologyPointDataVO;
import com.zhitan.ems.app.sse.SseEmitterUTF8;
import com.zhitan.ems.app.sse.SseTopologyServiceImpl;
import com.zhitan.ems.app.stat.service.impl.InfluxdbService;
import com.zhitan.ems.app.stat.vo.DeviceDataDateVo;
import com.zhitan.ems.common.core.controller.BaseController;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;

/**
 * 实时数据推送控制器
 * 
 * @author zhitan
 * @email
 * @date 2025-10-28 23:01:38
 */
@Slf4j
@Api(tags = "实时数据推送", value = "设备组态图")
@RestController
@RequestMapping("intra/iot/sse")
public class SseTopologyController extends BaseController {
	@Autowired
	private DeviceTopologyPointService deviceTopologyPointService;
	@Autowired
	private InfluxdbService influxdbService;
	@Value("${influx.bucket}")
	private String bucket;

	@Resource
	private SseTopologyServiceImpl sseTopologyService;

	// 使用线程池替代直接创建线程
	private static final ExecutorService executorService = Executors.newFixedThreadPool(10);

	@GetMapping("/stream")
	@ApiOperation("测试SSE流")
	public SseEmitter stream() {
		// 用于创建一个 SSE 连接对象
		SseEmitter emitter = new SseEmitter();
		// 在后台线程中模拟实时数据
		executorService.submit(() -> {
			try {
				List<TopologyPointDataVO> sseData = new ArrayList<TopologyPointDataVO>();
				Map<String, TopologyPointDataVO> curNodeMap=sseTopologyService.getTopologMap();
				 List<TopologyPointDataVO> listData=new ArrayList<TopologyPointDataVO>();
				 for (Map.Entry<String, TopologyPointDataVO> entry : curNodeMap.entrySet()) {
			             String table ="emsplus";
			     		 String start ="-1d";
			     		 String nodeKey=entry.getKey();
			     		TopologyPointDataVO preData=entry.getValue();
			     		 String iotNodeKey=entry.getKey();
			     		 if(nodeKey.contains("@")) {
			     			iotNodeKey=nodeKey.replace("@", "_");
			     		 }
			     		 try {
			     			 List<FluxTable> fluxTableList = influxdbService.getLastInfluxDataByDeviceCodePoint(bucket, start, null, table, iotNodeKey);
			             	 DeviceDataDateVo dataDateVo = InfluxdbService.getResultOneRowByTagProcess(fluxTableList,true); 
			             	 if(dataDateVo==null||dataDateVo.getValue()==null) {
			             		TopologyPointDataVO itemData=new TopologyPointDataVO();
			                	itemData.setNodeKey(nodeKey);
			                	itemData.setNodeValue("-1");
			                	sseTopologyService.setTopologMap(nodeKey,itemData);
			             	 }else {
			             		TopologyPointDataVO itemData=new TopologyPointDataVO();
			                	itemData.setNodeKey(nodeKey);
			                	itemData.setNodeValue(dataDateVo.getValue().toString());
			                	sseTopologyService.setTopologMap(nodeKey,itemData);
			                	if(preData!=null&&preData.getNodeValue()!=null&&!preData.getNodeValue().equalsIgnoreCase(itemData.getNodeValue())) {
			                		listData.add(itemData);
			                	}
			                	
			             	 }
						} catch (Exception e) {
						}
			        }
				 if(sseData.size()==0) {
					 	TopologyPointDataVO itemData = new TopologyPointDataVO();
						itemData.setNodeKey("nodeKey_no");
						itemData.setNodeValue("" + RandomUtil.randomDouble(10, 99));
						sseData.add(itemData);
				 }
				emitter.send(SseEmitter.event().name("message").data(JSONUtil.toJsonStr(sseData)));
				Thread.sleep(1000*30);
				// 数据发送完成后，关闭连接
				emitter.complete();
			} catch (IOException | InterruptedException e) {
				// 发生错误时，关闭连接并报错
				log.error("SSE stream error", e);
				emitter.completeWithError(e);
			}
		});
		return emitter;
	}

	/**
	 * 创建SSE连接
	 * 
	 * @param topologyId 拓扑ID
	 * @return SseEmitterUTF8 SSE连接对象
	 */
	@GetMapping("/createConnect/{topologyId}")
	@ApiOperation("创建SSE连接")
	public SseEmitterUTF8 createConnect(@PathVariable("topologyId") String topologyId) {
		if (StrUtil.isBlank(topologyId)) {
			log.info("sse注册信息不全");
			return null;
		}
		return sseTopologyService.createConnect(topologyId);
	}

	/**
	 * 关闭SSE连接
	 * 
	 * @param topologyId 拓扑ID
	 * @return Boolean 是否成功
	 */
	@GetMapping("/closeConnect/{topologyId}")
	@ApiOperation("关闭SSE连接")
	public Boolean closeConnect(@PathVariable("topologyId") String topologyId) {
		ThreadUtil.execute(() -> {
			sseTopologyService.closeConnectTopology(topologyId);
		});
		return true;
	}

	/**
	 * 获取连接统计信息
	 * 
	 * @return Map 连接统计信息
	 */
	@GetMapping("/stats")
	@ApiOperation("获取SSE连接统计信息")
	public Map<String, Object> getConnectionStats() {
		return sseTopologyService.getConnectionStats();
	}

	/**
	 * 发送心跳消息
	 * 
	 * @param topologyId 拓扑ID
	 * @return Boolean 是否成功
	 */
	@GetMapping("/heartbeat/{topologyId}")
	@ApiOperation("发送心跳消息")
	public Boolean sendHeartbeat(@PathVariable("topologyId") String topologyId) {
		if (StrUtil.isBlank(topologyId)) {
			return false;
		}
		sseTopologyService.sendHeartbeat(topologyId);
		return true;
	}
}